package fun.tan90.easy.binlog.manager.handler;

import com.mybatisflex.core.FlexGlobalConfig;
import com.taosdata.jdbc.SchemalessWriter;
import com.taosdata.jdbc.enums.SchemalessProtocolType;
import com.taosdata.jdbc.enums.SchemalessTimestampType;
import fun.tan90.easy.binlog.base.constants.EasyBinlogConstants;
import fun.tan90.easy.binlog.base.strategy.AbstractExecuteStrategy;
import fun.tan90.easy.binlog.base.threadpool.EasyBinlogThreadPool;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/12/15 19:42
 */
@Slf4j
@Component
public class MetricCollectHandler implements AbstractExecuteStrategy<String, Void> {

    private final DataSource dataSource = FlexGlobalConfig.getDefaultConfig().getDataSource().getDataSourceMap().get("tdengine");

    private final BlockingQueue<String> BLOCKING_QUEUE = new ArrayBlockingQueue<>(1024);

    public MetricCollectHandler() {
        List<String> schemalessSqlList = new ArrayList<>();
        EasyBinlogThreadPool.EB_SCHEDULED_EXECUTOR.scheduleWithFixedDelay(() -> {
            if (schemalessSqlList.isEmpty()) {
                if (BLOCKING_QUEUE.isEmpty()) {
                    return;
                }
                BLOCKING_QUEUE.drainTo(schemalessSqlList, Math.min(BLOCKING_QUEUE.size(), 50));
            }
            if (schemalessSqlList.isEmpty()) {
                return;
            }
            try {
                try (Connection connection = dataSource.getConnection();
                     SchemalessWriter writer = new SchemalessWriter(connection)) {
                    writer.write(schemalessSqlList, SchemalessProtocolType.LINE, SchemalessTimestampType.MILLI_SECONDS);
                }
            } catch (Exception e) {
                for (String schemalessSql : schemalessSqlList) {
                    if (!BLOCKING_QUEUE.offer(schemalessSql)) {
                        log.error("BlockingQueue offer failed [{}]", schemalessSql);
                    }
                }
            } finally {
                schemalessSqlList.clear();
            }
        }, 1, 60, TimeUnit.SECONDS);
    }

    @Override
    public String mark() {
        return EasyBinlogConstants.METRIC_COLLECT;
    }

    @Override
    public void execute(String schemalessSql) {
        if (!BLOCKING_QUEUE.offer(schemalessSql)) {
            log.error("BlockingQueue offer failed [{}]", schemalessSql);
        }
    }
}
